﻿using RockBus.DataContracts;
using RockBus.DataContracts.ConfigurationService;
using RockBus.Logging.Interfaces;
using RockBus.ConfigurationRepository.Interfaces;
using System;
using System.Collections.Generic;
using System.Configuration;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Transactions;

namespace RockBus.ConfigurationRepository.SqlServer
{
    public class DbConfigurationRepositoryConfiguration : IDbConfigurationRepositoryConfiguration
    {
        public DbConfigurationRepositoryConfiguration()
        {
            this.ConnectionString = ConfigurationManager.ConnectionStrings["RockBusDb"].ConnectionString;
        }

        public string ConnectionString { get; private set; }
    }

    public class ConfigurationRepository : IConfigurationRepository
    {
        public ConfigurationRepository(ILogger logger, IDbConfigurationRepositoryConfiguration configuration)
        {
            this.Logger = logger;
            this.Configuration = configuration;
        }

        public ILogger Logger { get; set; }

        public IDbConfigurationRepositoryConfiguration Configuration { get; set; }

        public void ClearAllClients()
        {
            this.RemoveAllSubscribers(new RemoveAllSubscribersRequest());
            this.RemoveAllPublishers(new RemoveAllPublishersRequest());
        }

        public void RemoveAllSubscribers(RemoveAllSubscribersRequest request)
        {
            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();
                SqlCommand sc = new SqlCommand("sp_ClearAllSubscribers", conn);
                sc.CommandType = CommandType.StoredProcedure;
                sc.ExecuteNonQuery();
                ts.Complete();
            }

            return;
        }

        public void RemoveAllPublishers(RemoveAllPublishersRequest request)
        {
            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();
                SqlCommand sc = new SqlCommand("sp_ClearAllPublishers", conn);
                sc.CommandType = CommandType.StoredProcedure;
                sc.ExecuteNonQuery();
                ts.Complete();
            }

            return;
        }

        public void AddOrUpdateSubscriberConfiguration(SubscriberConfiguration subscriberConfiguration)
        {
            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();

                foreach(SubscriberInfo si in subscriberConfiguration.EvaluationElementDictionary.Values.ToArray())
                {
                    // Subscriber
                    long subscriberDbId = SubscriberInsertUpdate(conn, si.Id);

                    // Namespaces
                    IDictionary<string, long> niIdMap = this.NamespaceDefinitionsInsertUpdate(conn, subscriberDbId, si.NamespaceInfoDictionary.Values);

                    // Filters
                    IDictionary<string, long> fiIdMap = this.FiltersInsertUpdate(conn, subscriberDbId, si.FilterInfoDictionary.Values);

                    // Endpoints
                    IDictionary<string, long> eiIdMap = this.EndpointsInsertUpdate(conn, subscriberDbId, si.EndpointInfoDictionary.Values);

                    // Pipelines
                    IDictionary<string, long> piIdMap = this.PipelinesInsertUpdate(conn, subscriberDbId, si.PipelineInfoDictionary.Values);

                    // SubscriptionInfos
                    foreach(KeyValuePair<string, SubscriptionInfo> kvp in si.InfoDictionary)
                    {
                        // Subscription
                        SubscriptionInfo sci = kvp.Value;
                        long subscriptionDbId = this.SubscriptionInsertUpdate(conn, subscriberDbId, sci);

                        // Subscription Filter
                        foreach(FilterInfo fi in sci.GetFilterInfoEnumerator())
                        {
                            long filterDbId = fiIdMap[fi.Id];
                            this.SubscriptionFilterInsertUpdate(conn, subscriptionDbId, filterDbId);
                        }
                    }
                }

                ts.Complete();
            }

            return;
        }

        private long SubscriberInsertUpdate(SqlConnection conn, string subscriberId)
        {
            SqlCommand sc = new SqlCommand("sp_Subscriber_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = subscriberId;
            sc.Parameters.Add("@ClientTypeDbId", SqlDbType.BigInt).Value = (long)ClientTypeEnum.Subscriber;

            SqlParameter subscriberDbIdParameter = new SqlParameter("@SubscriberDbId", 0);
            subscriberDbIdParameter.SqlDbType = SqlDbType.BigInt;
            subscriberDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(subscriberDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            return (long)subscriberDbIdParameter.Value;
        }

        private long SubscriptionInsertUpdate(SqlConnection conn, long subscriberDbId, SubscriptionInfo subscriptionInfo)
        {
            SqlCommand sc = new SqlCommand("sp_Subscription_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = subscriptionInfo.Detail.Id;
            sc.Parameters.Add("@Name", SqlDbType.NVarChar).Value = subscriptionInfo.Detail.Name;
            sc.Parameters.Add("@EndpointId", SqlDbType.NVarChar).Value = subscriptionInfo.Detail.EndpointId;
            sc.Parameters.Add("@Persistent", SqlDbType.Bit).Value = subscriptionInfo.Detail.Persistent;
            sc.Parameters.Add("@SubscriberDbId", SqlDbType.NVarChar).Value = subscriberDbId;
            sc.Parameters.Add("@Paused", SqlDbType.Bit).Value = subscriptionInfo.Detail.Paused;
            sc.Parameters.Add("@ExpiryDateTimeUtc", SqlDbType.DateTime2).Value = subscriptionInfo.Detail.ExpiryDateTimeUtc;
            sc.Parameters.Add("@MessageValidityDurationSecs", SqlDbType.BigInt).Value = subscriptionInfo.Detail.MessageValidityDurationSecs;
            sc.Parameters.Add("@DeliveryAttemptIntervalSecs", SqlDbType.Bit).Value = subscriptionInfo.Detail.DeliveryAttemptIntervalSecs;
            sc.Parameters.Add("@MaxDeliveryAttemptCount", SqlDbType.BigInt).Value = subscriptionInfo.Detail.MaxDeliveryAttemptCount;

            if(!string.IsNullOrEmpty(subscriptionInfo.Detail.RequestPipelineId))
            {
                sc.Parameters.Add("@RequestPipelineId", SqlDbType.NVarChar).Value = subscriptionInfo.Detail.RequestPipelineId;
            }
            if(!string.IsNullOrEmpty(subscriptionInfo.Detail.ResponsePipelineId))
            {
                sc.Parameters.Add("@ResponsePipelineId", SqlDbType.NVarChar).Value = subscriptionInfo.Detail.ResponsePipelineId;
            }

            SqlParameter subscriptionDbIdParameter = new SqlParameter("@SubscriptionDbId", 0);
            subscriptionDbIdParameter.SqlDbType = SqlDbType.BigInt;
            subscriptionDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(subscriptionDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            return (long)subscriptionDbIdParameter.Value;
        }

        private IDictionary<string, long> EndpointsInsertUpdate(SqlConnection conn, long clientDbId, IEnumerable<EndpointInfo> endpoints)
        {
            IDictionary<string, long> idMap = new Dictionary<string, long>();
            foreach(EndpointInfo ei in endpoints)
            {
                this.EndpointInsertUpdate(conn, clientDbId, ei, idMap);
            }
            return idMap;
        }

        private void EndpointInsertUpdate(SqlConnection conn, long? clientDbId, EndpointInfo ei, IDictionary<string, long> idMap)
        {
            SqlCommand sc = new SqlCommand("sp_Endpoint_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = ei.Id;
            sc.Parameters.Add("@ClientDbId", SqlDbType.BigInt).Value = clientDbId;
            sc.Parameters.Add("@Address", SqlDbType.NVarChar).Value = ei.Address;
            sc.Parameters.Add("@EndpointNameDbId", SqlDbType.BigInt).Value = (long)ei.EndpointName;
            sc.Parameters.Add("@CustomEndpointName", SqlDbType.NVarChar).Value = ei.CustomEndpointName;
            sc.Parameters.Add("@TransportTypeDbId", SqlDbType.BigInt).Value = (long)ei.TransportType;
            sc.Parameters.Add("@CustomTransportType", SqlDbType.NVarChar).Value = null;
            sc.Parameters.Add("@ExchangeTypeDbId", SqlDbType.BigInt).Value = (long)ei.ExchangeType;
            sc.Parameters.Add("@TransactionScopeOptionDbId", SqlDbType.BigInt).Value = (long)ei.TransactionScopeOption;

            SqlParameter endpointDbIdParameter = new SqlParameter("@EndpointDbId", 0);
            endpointDbIdParameter.SqlDbType = SqlDbType.BigInt;
            endpointDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(endpointDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            if (idMap != null)
            {
                idMap.Add(ei.Id, (long)endpointDbIdParameter.Value);
            }
        }

        public IDictionary<string, long> FiltersInsertUpdate(SqlConnection conn, long clientDbId, ICollection<FilterInfo> filters)
        {
            FilterInfo[] sortedFilters = filters.ToArray();
            Array.Sort<FilterInfo>(sortedFilters, new FilterInfoComparerForFilterReferences(true));

            IDictionary<string, long> idMap = new Dictionary<string, long>();
            foreach(FilterInfo fi in sortedFilters)
            {
                this.FilterInsertUpdate(conn, clientDbId, fi, idMap);
            }

            return idMap;
        }

        public long FilterInsertUpdate(SqlConnection conn, long clientDbId, FilterInfo fi, IDictionary<string, long> idMap)
        {
            SqlCommand sc = new SqlCommand("sp_Filter_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = fi.Id;
            sc.Parameters.Add("@ClientDbId", SqlDbType.BigInt).Value = clientDbId;
            sc.Parameters.Add("@FilterTypeDbId", SqlDbType.BigInt).Value = (long)fi.FilterType;
            sc.Parameters.Add("@FilterData", SqlDbType.NVarChar).Value = fi.FilterData;
            sc.Parameters.Add("@CustomTypeName", SqlDbType.NVarChar).Value = fi.CustomTypeName;
            sc.Parameters.Add("@Filter1DbId", SqlDbType.BigInt).Value = this.GetFilterDbId(fi.Filter1Id, idMap);
            sc.Parameters.Add("@Filter2DbId", SqlDbType.BigInt).Value = this.GetFilterDbId(fi.Filter2Id, idMap);

            SqlParameter filterDbIdParameter = new SqlParameter("@FilterDbId", 0);
            filterDbIdParameter.SqlDbType = SqlDbType.BigInt;
            filterDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(filterDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            idMap.Add(fi.Id, (long)filterDbIdParameter.Value);
            return (long)filterDbIdParameter.Value;
        }

        private void SubscriptionFilterInsertUpdate(SqlConnection conn, long subscriptionDbId, long filterDbId)
        {
            SqlCommand sc = new SqlCommand("sp_Subscription_Filter_InsertUpdate", conn);
            sc.Parameters.Add("@SubscriptionDbId", SqlDbType.BigInt).Value = subscriptionDbId;
            sc.Parameters.Add("@FilterDbId", SqlDbType.BigInt).Value = (long)filterDbId;
            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();
        }

        private object GetFilterDbId(string filterId, IDictionary<string, long> idMap)
        {
            if(string.IsNullOrEmpty(filterId))
            {
                return DBNull.Value;
            }

            return idMap[filterId];
        }

        private IDictionary<string, long> NamespaceDefinitionsInsertUpdate(SqlConnection conn, long clientDbId, IEnumerable<NamespaceInfo> namespaceDefinitions)
        {
            IDictionary<string, long> idMap = new Dictionary<string, long>();
            foreach(NamespaceInfo ni in namespaceDefinitions)
            {
                this.NamespaceDefinitionInsertUpdate(conn, clientDbId, ni, idMap);
            }
            return idMap;
        }

        private void NamespaceDefinitionInsertUpdate(SqlConnection conn, long clientDbId, NamespaceInfo ni, IDictionary<string, long> idMap)
        {
            SqlCommand sc = new SqlCommand("sp_NamespaceDefinition_InsertUpdate", conn);
            sc.Parameters.Add("@ClientDbId", SqlDbType.BigInt).Value = clientDbId;
            sc.Parameters.Add("@Prefix", SqlDbType.NVarChar).Value = ni.Prefix;
            sc.Parameters.Add("@Namespace", SqlDbType.NVarChar).Value = ni.Namespace;

            SqlParameter namespaceDefinitionDbIdParameter = new SqlParameter("@namespaceDefinitionDbId", 0);
            namespaceDefinitionDbIdParameter.SqlDbType = SqlDbType.BigInt;
            namespaceDefinitionDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(namespaceDefinitionDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            idMap.Add(ni.Prefix, (long)namespaceDefinitionDbIdParameter.Value);
        }

        public IDictionary<string, long> PipelinesInsertUpdate(SqlConnection conn, long clientDbId, ICollection<PipelineInfo> pipelines)
        {
            IDictionary<string, long> idMap = new Dictionary<string, long>();
            foreach(PipelineInfo pi in pipelines)
            {
                this.PipelineInsertUpdate(conn, clientDbId, pi, idMap);
            }

            return idMap;
        }

        public long PipelineInsertUpdate(SqlConnection conn, long clientDbId, PipelineInfo pi, IDictionary<string, long> idMap)
        {
            SqlCommand sc = new SqlCommand("sp_Pipeline_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = pi.Id;
            sc.Parameters.Add("@ClientDbId", SqlDbType.BigInt).Value = clientDbId;

            var pipelineInfoByteArray = pi.GetByteArray();
            sc.Parameters.Add("@PipelineInfo", SqlDbType.VarBinary).Value = pipelineInfoByteArray;

            SqlParameter pipelineDbIdParameter = new SqlParameter("@PipelineDbId", 0);
            pipelineDbIdParameter.SqlDbType = SqlDbType.BigInt;
            pipelineDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(pipelineDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            long piDbId = (long)pipelineDbIdParameter.Value;
            idMap.Add(pi.Id, piDbId);

            return (long)pipelineDbIdParameter.Value;
        }

        public void RemoveSubscribers(RemoveSubscribersRequest request)
        {
            var clientIds = new DataTable();
            clientIds.Columns.Add("Id", typeof(string));
            foreach(string subscriberId in request.SubscriberIds)
            {
                clientIds.Rows.Add(subscriberId);
            }

            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
                {
                conn.Open();

                    SqlCommand sc = new SqlCommand("sp_Client_Delete", conn);
                sc.Parameters.Add("@ClientIds", SqlDbType.Structured).Value = clientIds;
                    sc.CommandType = CommandType.StoredProcedure;
                    sc.ExecuteNonQuery();

                ts.Complete(); // D'OH!
                }
            }

        public void RemoveSubscriptions(RemoveSubscriptionsRequest request)
        {
            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();
                foreach(string subscriptionId in request.SubscriptionIds)
                {
                    this.RemoveSubscription(conn, request.SubscriberId, subscriptionId);
                }

                ts.Complete(); // D'OH!
        }
        }

        private void RemoveSubscription(SqlConnection conn, string subscriberId, string subscriptionId)
        {
            SqlCommand sc = new SqlCommand("sp_Subscription_Delete", conn);
            sc.Parameters.Add("@ClientId", SqlDbType.NVarChar).Value = subscriberId;
            sc.Parameters.Add("@SubscriptionId", SqlDbType.NVarChar).Value = subscriptionId;
            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();
        }

        public SubscriberConfiguration GetSubscriberConfiguration()
        {
            SubscriberConfiguration sc = new SubscriberConfiguration();

            IDictionary<string, long> idMap = new Dictionary<string, long>();
            using(SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();

                SqlCommand cmd = new SqlCommand("sp_Subscriber_Select", conn);
                cmd.CommandType = CommandType.StoredProcedure;
                using(SqlDataReader reader = cmd.ExecuteReader())
                {
                    while(reader.Read())
                    {
                        long dbId = (long)reader["DbId"];
                        SubscriberInfo si = new SubscriberInfo(
                            (string)reader["Id"],
                            modificationDateTimeUtc: (DateTime)reader["ModificationDateTimeUtc"]
                            );

                        idMap.Add(si.Id, dbId);
                        sc.EvaluationElementDictionary.Add(si.Id, si);
                    }
                }

                foreach(SubscriberInfo si in sc.EvaluationElementDictionary.Values)
                {
                    long dbId = idMap[si.Id];

                    // Namespaces
                    si.NamespaceInfoDictionary = this.ReadNamespaces(conn, dbId);

                    // Filters
                    si.FilterInfoDictionary = this.ReadFilters(conn, dbId);

                    // Pipelines
                    si.PipelineInfoDictionary = this.ReadPipelines(conn, dbId);

                    // Endpoints
                    si.EndpointInfoDictionary = this.ReadEndpoints(conn, dbId);

                    // Subscriptions
                    si.InfoDictionary = this.ReadSubscriptions(conn, dbId, si);
                }
            }

            return sc;
        }

        private CaseInsensitiveStringDictionary<NamespaceInfo> ReadNamespaces(SqlConnection conn, long clientDbId)
        {
            var namespaceDictionary = new CaseInsensitiveStringDictionary<NamespaceInfo>();

            SqlCommand cmd = new SqlCommand("sp_NamespaceDefinition_Select", conn);
            cmd.Parameters.Add("@clientDbId", SqlDbType.NVarChar).Value = clientDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                while(reader.Read())
                {
                    long dbId = (long)reader["DbId"];
                    var ni = new NamespaceInfo()
                    {
                        Prefix = (string)reader["Prefix"],
                        Namespace = (string)reader["Namespace"],
                    };

                    namespaceDictionary.Add(ni.Prefix, ni);
                }
            }

            return namespaceDictionary;
        }

        private CaseInsensitiveStringDictionary<FilterInfo> ReadFilters(SqlConnection conn, long clientDbId)
        {
            var filterDictionary = new CaseInsensitiveStringDictionary<FilterInfo>();

            SqlCommand cmd = new SqlCommand("sp_Filter_Select", conn);
            cmd.Parameters.Add("@clientDbId", SqlDbType.NVarChar).Value = clientDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                while(reader.Read())
                {
                    long dbId = (long)reader["DbId"];
                    FilterInfo fi = new FilterInfo()
                    {
                        Id = (string)reader["Id"],
                        FilterType = (FilterTypeEnum)(long)reader["FilterTypeDbId"],
                        FilterData = reader.GetFromNullableColumn<string>("FilterData"),
                        CustomTypeName = reader.GetFromNullableColumn<string>("CustomTypeName"),
                        Filter1Id = reader.GetFromNullableColumn<string>("Filter1Id"),
                        Filter2Id = reader.GetFromNullableColumn<string>("Filter2Id"),
                    };

                    filterDictionary.Add(fi.Id, fi);
                }
            }

            return filterDictionary;
        }

        private CaseInsensitiveStringDictionary<PipelineInfo> ReadPipelines(SqlConnection conn, long clientDbId)
        {
            var pipelineDictionary = new CaseInsensitiveStringDictionary<PipelineInfo>();

            SqlCommand cmd = new SqlCommand("sp_Pipeline_Select", conn);
            cmd.Parameters.Add("@clientDbId", SqlDbType.NVarChar).Value = clientDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                // PipelineDbId, PipelineId, PipelineInfo
                while(reader.Read())
                {
                    byte[] pipelineInfoByteArray = (byte[])reader["PipelineInfo"];
                    var pi = PipelineInfo.CreateFromByteArray(pipelineInfoByteArray);

                    pipelineDictionary.Add(pi.Id, pi);
                }
            }

            return pipelineDictionary;
        }

        private CaseInsensitiveStringDictionary<EndpointInfo> ReadEndpoints(SqlConnection conn, long clientDbId)
        {
            var endpointDictionary = new CaseInsensitiveStringDictionary<EndpointInfo>();

            SqlCommand cmd = new SqlCommand("sp_Endpoint_Select", conn);
            cmd.Parameters.Add("@ClientDbId", SqlDbType.BigInt).Value = clientDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                while(reader.Read())
                {
                    long dbId = (long)reader["DbId"];
                    EndpointInfo ei = new EndpointInfo()
                    {
                        Id = (string)reader["Id"],
                        Address = (string)reader["Address"],
                        ExchangeType = (ExchangeTypeEnum)(long)reader["ExchangeTypeDbId"],
                        EndpointName = (EndpointNameEnum)(long)reader["EndpointNameDbId"],
                        CustomEndpointName = reader.GetFromNullableColumn<string>("CustomEndpointName"),
                        TransportType = (TransportTypeEnum)(long)reader["TransportTypeDbId"],
                        TransactionScopeOption = (TransactionScopeOption)(long)reader["TransactionScopeOptionDbId"],
                    };

                    endpointDictionary.Add(ei.Id, ei);
                }
            }

            return endpointDictionary;
        }

        private EndpointInfo ReadEndpoint(SqlConnection conn, string endpointId)
        {
            SqlCommand cmd = new SqlCommand("sp_Endpoint_SelectByEndpointId", conn);
            cmd.Parameters.Add("@EndpointId", SqlDbType.NVarChar).Value = endpointId;
            cmd.CommandType = CommandType.StoredProcedure;
            using (SqlDataReader reader = cmd.ExecuteReader())
            {
                if (reader.Read())
                {
                    long dbId = (long)reader["DbId"];
                    EndpointInfo ei = new EndpointInfo()
                    {
                        Id = (string)reader["Id"],
                        Address = (string)reader["Address"],
                        ExchangeType = (ExchangeTypeEnum)(long)reader["ExchangeTypeDbId"],
                        EndpointName = (EndpointNameEnum)(long)reader["EndpointNameDbId"],
                        CustomEndpointName = reader.GetFromNullableColumn<string>("CustomEndpointName"),
                        TransportType = (TransportTypeEnum)(long)reader["TransportTypeDbId"],
                        TransactionScopeOption = (TransactionScopeOption)(long)reader["TransactionScopeOptionDbId"],
                    };

                    return ei;
                }
            }

            return null;
        }

        private CaseInsensitiveStringDictionary<SubscriptionInfo> ReadSubscriptions(SqlConnection conn, long subscriberDbId, SubscriberInfo si)
        {
            var subscriptionDetailDictionary = new CaseInsensitiveStringDictionary<SubscriptionDetail>();

            IDictionary<string, long> idMap = new Dictionary<string, long>();

            SqlCommand cmd = new SqlCommand("sp_Subscription_Select", conn);
            cmd.Parameters.Add("@SubscriberDbId", SqlDbType.NVarChar).Value = subscriberDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                while(reader.Read())
                {
                    long dbId = (long)reader["DbId"];

                    var requestPipelineId = reader.GetFromNullableColumn<string>("RequestPipelineId");
                    var responsePipelineId = reader.GetFromNullableColumn<string>("ResponsePipelineId");

                    SubscriptionDetail sd = new SubscriptionDetail(
                        si.Id,
                        (string)reader["Id"],
                        reader.GetFromNullableColumn<string>("Name"),
                        (string)reader["EndpointId"],
                        (bool)reader["Persistent"],
                        reader.GetFromNullableColumn<long?>("MessageValidityDurationSecs"),
                        reader.GetFromNullableColumn<long?>("DeliveryAttemptIntervalSecs"),
                        reader.GetFromNullableColumn<int>("MaxDeliveryAttemptCount"),
                        (bool)reader["Paused"],
                        reader.GetFromNullableColumn<DateTime?>("ExpiryDateTimeUtc"),
                        requestPipelineId,
                        responsePipelineId
                        );

                    idMap.Add(sd.Id, dbId);
                    subscriptionDetailDictionary.Add(sd.Id, sd);
                }
            }

            var subscriptionDictionary = new CaseInsensitiveStringDictionary<SubscriptionInfo>();
            foreach(SubscriptionDetail sd in subscriptionDetailDictionary.Values)
            {
                long dbId = idMap[sd.Id];

                var fis = this.ReadSubscriptionFilters(conn, dbId);

                SubscriptionInfo sci = new SubscriptionInfo(si, sd, fis);
                subscriptionDictionary.Add(sci.Detail.Id, sci);
            }

            return subscriptionDictionary;
        }

        private IEnumerable<string> ReadSubscriptionFilters(SqlConnection conn, long subscriptionDbId)
        {
            IList<string> subscriptionFilters = new List<string>();

            SqlCommand cmd = new SqlCommand("sp_Subscription_Filter_Select", conn);
            cmd.Parameters.Add("@SubscriptionDbId", SqlDbType.NVarChar).Value = subscriptionDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                while(reader.Read())
                {
                    string filterId = (string)reader["FilterId"];
                    subscriptionFilters.Add(filterId);
                }
            }

            return subscriptionFilters;
        }

        public void AddOrUpdatePublisherConfiguration(PublisherConfiguration publisherConfiguration)
        {
            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();

                foreach(PublisherInfo pi in publisherConfiguration.EvaluationElementDictionary.Values.ToArray())
                {
                    // Publisher
                    long publisherDbId = this.PublisherInsertUpdate(conn, pi.Id);

                    // Endpoints
                    IDictionary<string, long> eiIdMap = this.EndpointsInsertUpdate(conn, publisherDbId, pi.EndpointInfoDictionary.Values);

                    // Pipelines
                    IDictionary<string, long> piIdMap = this.PipelinesInsertUpdate(conn, publisherDbId, pi.PipelineInfoDictionary.Values);

                    // Responses
                    IDictionary<string, long> riIdMap = this.ResponsesInsertUpdate(conn, publisherDbId, pi.ResponseInfoDictionary.Values);

                    // ReceivePortInfos
                    foreach(KeyValuePair<string, ReceivePortInfo> kvp in pi.InfoDictionary)
                    {
                        // ReceivePort
                        ReceivePortInfo pci = kvp.Value;
                        long receivePortDbId = this.ReceivePortInsertUpdate(conn, publisherDbId, pci);
                    }
                }

                ts.Complete();
            }

            return;
        }

        private long PublisherInsertUpdate(SqlConnection conn, string publisherId)
        {
            SqlCommand sc = new SqlCommand("sp_Publisher_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = publisherId;
            sc.Parameters.Add("@ClientTypeDbId", SqlDbType.BigInt).Value = (long)ClientTypeEnum.Publisher;

            SqlParameter subscriberDbIdParameter = new SqlParameter("@PublisherDbId", 0);
            subscriberDbIdParameter.SqlDbType = SqlDbType.BigInt;
            subscriberDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(subscriberDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            return (long)subscriberDbIdParameter.Value;
        }

        private long ReceivePortInsertUpdate(SqlConnection conn, long publisherDbId, ReceivePortInfo receivePortInfo)
        {
            SqlCommand sc = new SqlCommand("sp_ReceivePort_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = receivePortInfo.Detail.Id;
            sc.Parameters.Add("@Name", SqlDbType.NVarChar).Value = receivePortInfo.Detail.Name;
            sc.Parameters.Add("@Paused", SqlDbType.Bit).Value = receivePortInfo.Detail.Paused;
            sc.Parameters.Add("@PublisherDbId", SqlDbType.NVarChar).Value = publisherDbId;
            sc.Parameters.Add("@ExpiryDateTimeUtc", SqlDbType.DateTime2).Value = receivePortInfo.Detail.ExpiryDateTimeUtc;
            sc.Parameters.Add("@GenerateDefaultResponse", SqlDbType.Bit).Value = receivePortInfo.GenerateDefaultResponse;

            if(!string.IsNullOrEmpty(receivePortInfo.Detail.EndpointId))
            {
                sc.Parameters.Add("@EndpointId", SqlDbType.NVarChar).Value = receivePortInfo.Detail.EndpointId;
            }

            if(!string.IsNullOrEmpty(receivePortInfo.Detail.RequestPipelineId))
            {
                sc.Parameters.Add("@RequestPipelineId", SqlDbType.NVarChar).Value = receivePortInfo.Detail.RequestPipelineId;
            }
            if(!string.IsNullOrEmpty(receivePortInfo.Detail.ResponsePipelineId))
            {
                sc.Parameters.Add("@ResponsePipelineId", SqlDbType.NVarChar).Value = receivePortInfo.Detail.ResponsePipelineId;
            }

            SqlParameter receivePortDbIdParameter = new SqlParameter("@ReceivePortDbId", 0);
            receivePortDbIdParameter.SqlDbType = SqlDbType.BigInt;
            receivePortDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(receivePortDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            return (long)receivePortDbIdParameter.Value;
        }

        private IDictionary<string, long> ResponsesInsertUpdate(SqlConnection conn, long publisherDbId, IEnumerable<ResponseInfo> responses)
        {
            IDictionary<string, long> idMap = new Dictionary<string, long>();
            foreach(ResponseInfo ri in responses)
            {
                this.ResponseInsertUpdate(conn, publisherDbId, ri, idMap);
            }

            return idMap;
        }

        private long ResponseInsertUpdate(SqlConnection conn, long publisherDbId, ResponseInfo ri, IDictionary<string, long> idMap)
        {
            SqlCommand sc = new SqlCommand("sp_Response_InsertUpdate", conn);
            sc.Parameters.Add("@Id", SqlDbType.NVarChar).Value = ri.Id;
            sc.Parameters.Add("@PublisherDbId", SqlDbType.BigInt).Value = publisherDbId;
            sc.Parameters.Add("@ResponseAction", SqlDbType.NVarChar).Value = (string)ri.ResponseAction;
            sc.Parameters.Add("@BufferSize", SqlDbType.Int).Value = ri.BufferSize;
            sc.Parameters.Add("@ResponseXml", SqlDbType.NVarChar).Value = ri.ResponseXml;

            SqlParameter responseDbIdParameter = new SqlParameter("@ResponseDbId", 0);
            responseDbIdParameter.SqlDbType = SqlDbType.BigInt;
            responseDbIdParameter.Direction = ParameterDirection.Output;
            sc.Parameters.Add(responseDbIdParameter);

            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();

            idMap.Add(ri.Id, (long)responseDbIdParameter.Value);
            return (long)responseDbIdParameter.Value;
        }

        public void RemovePublishers(RemovePublishersRequest request)
        {
            var clientIds = new DataTable();
            clientIds.Columns.Add("Id", typeof(string));
            foreach(string publisherId in request.PublisherIds)
            {
                clientIds.Rows.Add(publisherId);
            }

            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();

                    SqlCommand sc = new SqlCommand("sp_Client_Delete", conn);
                sc.Parameters.Add("@ClientIds", SqlDbType.Structured).Value = clientIds;
                    sc.CommandType = CommandType.StoredProcedure;
                    sc.ExecuteNonQuery();

                ts.Complete(); // D'OH!
            }
        }

        public void RemoveReceivePorts(RemoveReceivePortsRequest request)
        {
            using(var ts = new TransactionScope())
            using(var conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();
                foreach(string ReceivePortId in request.ReceivePortIds)
                {
                    this.RemoveReceivePort(conn, request.PublisherId, ReceivePortId);
                }

                ts.Complete(); // D'OH!
            }

            return;
        }

        private void RemoveReceivePort(SqlConnection conn, string publisherId, string receivePortId)
        {
            SqlCommand sc = new SqlCommand("sp_ReceivePort_Delete", conn);
            sc.Parameters.Add("@ClientId", SqlDbType.NVarChar).Value = publisherId;
            sc.Parameters.Add("@ReceivePortId", SqlDbType.NVarChar).Value = receivePortId;
            sc.CommandType = CommandType.StoredProcedure;
            sc.ExecuteNonQuery();
        }

        private CaseInsensitiveStringDictionary<ReceivePortInfo> ReadReceivePorts(SqlConnection conn, long publisherDbId, PublisherInfo pi)
        {
            var receivePortDictionary = new CaseInsensitiveStringDictionary<ReceivePortInfo>();

            IDictionary<string, long> idMap = new Dictionary<string, long>();

            SqlCommand cmd = new SqlCommand("sp_ReceivePort_Select", conn);
            cmd.Parameters.Add("@PublisherDbId", SqlDbType.NVarChar).Value = publisherDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using(SqlDataReader reader = cmd.ExecuteReader())
            {
                while(reader.Read())
                {
                    long dbId = (long)reader["DbId"];

                    var endpointId = reader.GetFromNullableColumn<string>("EndpointId");
                    var requestPipelineId = reader.GetFromNullableColumn<string>("RequestPipelineId");
                    var responsePipelineId = reader.GetFromNullableColumn<string>("ResponsePipelineId");

                    ReceivePortDetail pd = new ReceivePortDetail(
                        pi.Id,
                        (string)reader["Id"],
                        reader.GetFromNullableColumn<string>("Name"),
                        endpointId,
                        (bool)reader["Paused"],
                        reader.GetFromNullableColumn<DateTime?>("ExpiryDateTimeUtc"),
                        requestPipelineId,
                        responsePipelineId
                        );

                    var responseId = reader.GetFromNullableColumn<string>("ResponseId");
                    ReceivePortInfo ai = new ReceivePortInfo(
                        pi,
                        pd,
                        responseId,
                        (bool)reader["GenerateDefaultResponse"]);
                    idMap.Add(ai.Detail.Id, dbId);
                    receivePortDictionary.Add(ai.Detail.Id, ai);
                }
            }

            return receivePortDictionary;
        }

        public PublisherConfiguration GetPublisherConfiguration()
        {
            PublisherConfiguration pc = new PublisherConfiguration();

            IDictionary<string, long> idMap = new Dictionary<string, long>();

            using(SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();

                SqlCommand cmd = new SqlCommand("sp_Publisher_Select", conn);
                cmd.CommandType = CommandType.StoredProcedure;
                using(SqlDataReader reader = cmd.ExecuteReader())
                {
                    while(reader.Read())
                    {
                        long dbId = (long)reader["DbId"];
                        PublisherInfo pi = new PublisherInfo(
                            (string)reader["Id"],
                            modificationDateTimeUtc: (DateTime)reader["ModificationDateTimeUtc"]
                        );

                        idMap.Add(pi.Id, dbId);
                        pc.EvaluationElementDictionary.Add(pi.Id, pi);
                    }
                }

                foreach(PublisherInfo pi in pc.EvaluationElementDictionary.Values)
                {
                    long dbId = idMap[pi.Id];

                    // Endpoints
                    pi.EndpointInfoDictionary = this.ReadEndpoints(conn, dbId);

                    // Pipelines
                    pi.PipelineInfoDictionary = this.ReadPipelines(conn, dbId);

                    // Responses
                    pi.ResponseInfoDictionary = this.ReadPublisherResponses(conn, dbId);

                    // ReceivePorts
                    pi.InfoDictionary = this.ReadReceivePorts(conn, dbId, pi);
                }
            }

            return pc;
        }

        public SubscriberInfo GetSubscriber(GetSubscriberRequest request)
        {
            string sqlCommand = "select DbId, Id, ModificationDateTimeUtc from Client c where c.ClientTypeDbId = 0";

            SubscriberInfo si = null;
            using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                long dbId = 0;
                conn.Open();

                SqlCommand cmd = new SqlCommand();
                cmd.Connection = conn;
                cmd.CommandType = CommandType.Text;
                if (!string.IsNullOrEmpty(request.SubscriberId))
                {
                    cmd.Parameters.Add("@subscriptionId", SqlDbType.VarChar).Value = request.SubscriberId;
                    sqlCommand += " and c.Id = @subscriptionId";
                }
                cmd.CommandText = sqlCommand;

                using (SqlDataReader reader = cmd.ExecuteReader())
                {

                    while (reader.Read())
                    {
                        dbId = (long)reader["DbId"];
                        si = new SubscriberInfo(
                            (string)reader["Id"],
                            modificationDateTimeUtc: (DateTime)reader["ModificationDateTimeUtc"]
                            );
                    }
                }
                if (si == null)
                    return null;

                // Namespaces
                si.NamespaceInfoDictionary = this.ReadNamespaces(conn, dbId);

                // Filters
                si.FilterInfoDictionary = this.ReadFilters(conn, dbId);

                // Transformations
                si.PipelineInfoDictionary = this.ReadPipelines(conn, dbId);
               
                // Endpoints
                si.EndpointInfoDictionary = this.ReadEndpoints(conn, dbId);

                // Subscriptions
                si.InfoDictionary = this.ReadSubscriptions(conn, dbId, si);
            }

            return si;
        }

        private CaseInsensitiveStringDictionary<ResponseInfo> ReadPublisherResponses(SqlConnection conn, long publisherDbId)
        {
            var responseDictionary = new CaseInsensitiveStringDictionary<ResponseInfo>();

            SqlCommand cmd = new SqlCommand("sp_Response_Select", conn);
            cmd.Parameters.Add("@publisherDbId", SqlDbType.NVarChar).Value = publisherDbId;
            cmd.CommandType = CommandType.StoredProcedure;
            using (SqlDataReader reader = cmd.ExecuteReader())
            {
                while (reader.Read())
                {
                    long dbId = (long)reader["DbId"];
                    ResponseInfo ri = new ResponseInfo(
                        (string)reader["Id"],
                        (string)reader["ResponseAction"],
                        (string)reader["ResponseXml"],
                        (int)reader["BufferSize"]);

                    responseDictionary.Add(ri.Id, ri);
                }
            }

            return responseDictionary;
        }

        public InternalConfiguration GetInternalConfiguration()
        {
            InternalConfiguration configuration = new InternalConfiguration();

            using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                string requestPersistenceEndpointId = null;
                string responsePersistenceEndpointId = null;

                conn.Open();
                SqlCommand cmd = new SqlCommand("sp_InternalConfiguration_Select", conn);
                cmd.CommandType = CommandType.StoredProcedure;
                using (SqlDataReader reader = cmd.ExecuteReader())
                {
                    if (reader.Read())
                    {
                        configuration.CorrelationTimeoutMs = (int)reader["CorrelationTimeoutMs"];
                        configuration.DebugMessageLoggingEnabled = (bool)reader["DebugMessageLoggingEnabled"];
                        configuration.DefaultMaxMessageBufferSize = (int)reader["DefaultMaxMessageBufferSize"];
                        configuration.DefaultTransientDeliveryAttemptIntervalSecs = (int)reader["DefaultTransientDeliveryAttemptIntervalSecs"];
                        configuration.DefaultTransientMaxDeliveryAttemptCount = (int)reader["DefaultTransientMaxDeliveryAttemptCount"];
                        configuration.IsolationLevel = (IsolationLevelEnum)reader["IsolationLevel"];
                        configuration.MaxDegreeOfParallelism = (int)reader["MaxDegreeOfParallelism"];
                        configuration.MessageTimeoutMs = (int)reader["MessageTimeoutMs"];
                        configuration.PerformanceCounterUpdateIntervalMs = (int)reader["PerformanceCounterUpdateIntervalMs"];
                        configuration.ReinitializationIntervalMs = (int)reader["ReinitializationIntervalMs"];
                        configuration.ReinitLockTimeoutMs = (int)reader["ReinitLockTimeoutMs"];
                        configuration.ServiceHostOpenTimeoutMs = (int)reader["ServiceHostOpenTimeoutMs"];
                        configuration.TransactionTimeoutMs = (int)reader["TransactionTimeoutMs"];
                        configuration.BufferManagerMaxPoolSize = (int)reader["BufferManagerMaxPoolSize"];
                        configuration.BufferManagerMaxBufferSize = (int)reader["BufferManagerMaxBufferSize"];
                        configuration.MessageFragmentSize = (int)reader["MessageFragmentSize"];
                        configuration.PollingBatchSize = (int)reader["PollingBatchSize"];
                        configuration.PollingIntervalMs = (int)reader["PollingIntervalMs"];
                        configuration.DeliveryTimeoutMs = (int)reader["DeliveryTimeoutMs"];
                        configuration.DefaultPersistentDeliveryAttemptIntervalSecs = (int)reader["DefaultPersistentDeliveryAttemptIntervalSecs"];
                        configuration.DefaultPersistentMaxDeliveryAttemptCount = (int)reader["DefaultPersistentMaxDeliveryAttemptCount"];
                        configuration.ContentFolder = reader.GetFromNullableColumn<string>("ContentFolder");
                        configuration.CustomAssemblyFolder = reader.GetFromNullableColumn<string>("CustomAssemblyFolder");
                        requestPersistenceEndpointId = reader.GetFromNullableColumn<string>("RequestPersistenceEndpointId");
                        responsePersistenceEndpointId = reader.GetFromNullableColumn<string>("ResponsePersistenceEndpointId");
                    }
                }

                if (!string.IsNullOrEmpty(requestPersistenceEndpointId))
                {
                    configuration.RequestPersistenceEndpoint = this.ReadEndpoint(conn, requestPersistenceEndpointId);
                }
                if (!string.IsNullOrEmpty(responsePersistenceEndpointId))
                {
                    configuration.ResponsePersistenceEndpoint = this.ReadEndpoint(conn, responsePersistenceEndpointId);
                }
            }

            return configuration;
        }

        public void SetInternalConfiguration(InternalConfiguration configuration)
        {
            using (SqlConnection conn = new SqlConnection(this.Configuration.ConnectionString))
            {
                conn.Open();
                SqlCommand sc = new SqlCommand("sp_InternalConfiguration_InsertUpdate", conn);

                if (configuration.RequestPersistenceEndpoint != null)
                {
                    this.EndpointInsertUpdate(conn, null, configuration.RequestPersistenceEndpoint, null);
                    sc.Parameters.Add("@RequestPersistenceEndpointId", SqlDbType.NVarChar).Value = configuration.RequestPersistenceEndpoint.Id;
                }

                if (configuration.ResponsePersistenceEndpoint != null)
                {
                    this.EndpointInsertUpdate(conn, null, configuration.ResponsePersistenceEndpoint, null);
                    sc.Parameters.Add("@ResponsePersistenceEndpointId", SqlDbType.NVarChar).Value = configuration.ResponsePersistenceEndpoint.Id;
                }

                sc.Parameters.Add("@CorrelationTimeoutMs", SqlDbType.Int).Value = configuration.CorrelationTimeoutMs;
                sc.Parameters.Add("@DebugMessageLoggingEnabled", SqlDbType.Bit).Value = configuration.DebugMessageLoggingEnabled;
                sc.Parameters.Add("@DefaultMaxMessageBufferSize", SqlDbType.Int).Value = configuration.DefaultMaxMessageBufferSize;
                sc.Parameters.Add("@DefaultTransientDeliveryAttemptIntervalSecs", SqlDbType.Int).Value = configuration.DefaultTransientDeliveryAttemptIntervalSecs;
                sc.Parameters.Add("@DefaultTransientMaxDeliveryAttemptCount", SqlDbType.Int).Value = configuration.DefaultTransientMaxDeliveryAttemptCount;
                sc.Parameters.Add("@IsolationLevel", SqlDbType.Int).Value = (int) configuration.IsolationLevel;
                sc.Parameters.Add("@MaxDegreeOfParallelism", SqlDbType.Int).Value = configuration.MaxDegreeOfParallelism;
                sc.Parameters.Add("@MessageTimeoutMs", SqlDbType.Int).Value = configuration.MessageTimeoutMs;
                sc.Parameters.Add("@PerformanceCounterUpdateIntervalMs", SqlDbType.Int).Value = configuration.PerformanceCounterUpdateIntervalMs;
                sc.Parameters.Add("@ReinitializationIntervalMs", SqlDbType.Int).Value = configuration.ReinitializationIntervalMs;
                sc.Parameters.Add("@ReinitLockTimeoutMs", SqlDbType.Int).Value = configuration.ReinitLockTimeoutMs;
                sc.Parameters.Add("@ServiceHostOpenTimeoutMs", SqlDbType.Int).Value = configuration.ServiceHostOpenTimeoutMs;
                sc.Parameters.Add("@TransactionTimeoutMs", SqlDbType.Int).Value = configuration.TransactionTimeoutMs;
                sc.Parameters.Add("@BufferManagerMaxPoolSize", SqlDbType.Int).Value = configuration.BufferManagerMaxPoolSize;
                sc.Parameters.Add("@BufferManagerMaxBufferSize", SqlDbType.Int).Value = configuration.BufferManagerMaxBufferSize;
                sc.Parameters.Add("@MessageFragmentSize", SqlDbType.Int).Value = configuration.MessageFragmentSize;
                sc.Parameters.Add("@PollingBatchSize", SqlDbType.Int).Value = configuration.PollingBatchSize;
                sc.Parameters.Add("@PollingIntervalMs", SqlDbType.Int).Value = configuration.PollingIntervalMs;
                sc.Parameters.Add("@DeliveryTimeoutMs", SqlDbType.Int).Value = configuration.DeliveryTimeoutMs;
                sc.Parameters.Add("@DefaultPersistentDeliveryAttemptIntervalSecs", SqlDbType.Int).Value = configuration.DefaultPersistentDeliveryAttemptIntervalSecs;
                sc.Parameters.Add("@DefaultPersistentMaxDeliveryAttemptCount", SqlDbType.Int).Value = configuration.DefaultPersistentMaxDeliveryAttemptCount;
                sc.Parameters.Add("@ContentFolder", SqlDbType.NVarChar).Value = configuration.ContentFolder;
                sc.Parameters.Add("@CustomAssemblyFolder", SqlDbType.NVarChar).Value = configuration.CustomAssemblyFolder;
                sc.Parameters.Add("@LastModified", SqlDbType.DateTime2).Value = DateTime.UtcNow;
                sc.CommandType = CommandType.StoredProcedure;
                sc.ExecuteNonQuery();
            }
        }
    }

    public static class SqlDataReaderExtentions
    {
        public static T GetFromNullableColumn<T>(this SqlDataReader reader, string columnName)
        {
            if((reader == null) ||
                (reader[columnName] == null) ||
                (reader[columnName] == DBNull.Value))
            {
                return default(T);
            }

            return (T)reader[columnName];
        }
    }
}
